/*
 * Copyright 2014 Semidata. All rights reserved.
 */
package com.semidata.rtd.core.tcp.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.GenericFutureListener;

import java.util.concurrent.locks.ReentrantLock;

/**
 * @author shewei@semidata.com, She Wei
 * 
 */
public class TcpConnection {

	private NioEventLoopGroup group;
	private String hostname;
	private int port;
	private TcpResponseHandler handler;
	private Channel channel;
	private ReentrantLock lock;

	public TcpConnection(NioEventLoopGroup group, String hostname, int port,
	        TcpResponseHandler handler) {
		this.group = group;
		this.hostname = hostname;
		this.port = port;
		this.handler = handler;
		this.channel = null;
		this.lock = new ReentrantLock();
	}

	public boolean isActive() {
		return channel != null && channel.isActive() && channel.isOpen();
	}

	public void reconnect() {
		disconnect();
		connect();
	}

	public void disconnect() {
		disconnect(null);
	}

	public void disconnect(final FutureCompleteHandler disconnectCompleteHandler) {
		lock.lock();
		try {
			if (channel == null)
				return;

			ChannelFuture future = channel.disconnect();

			if (disconnectCompleteHandler != null) {
				future.addListener(new GenericFutureListener<ChannelFuture>() {
					@Override
					public void operationComplete(ChannelFuture future)
					        throws Exception {
						if (disconnectCompleteHandler != null)
							disconnectCompleteHandler.handle(future);
					}
				});
			} else {
				try {
					future.sync();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			channel = null;
		} finally {
			lock.unlock();
		}
	}

	public void connect() {
		connect(null);
	}

	public void connect(final FutureCompleteHandler connectCompleteHandler) {
		lock.lock();
		try {
			if (channel != null)
				return;

			Bootstrap b = new Bootstrap();
			b.group(group).channel(NioSocketChannel.class);
			b.handler(new TcpClientIntializer(handler));
			b.option(ChannelOption.SO_KEEPALIVE, true);
			try {
				if (connectCompleteHandler == null) {
					channel = b.connect(hostname, port).sync().channel();
				} else {

					b.connect(hostname, port).addListener(
					        new GenericFutureListener<ChannelFuture>() {
						        @Override
						        public void operationComplete(
						                ChannelFuture future) throws Exception {
							        channel = future.channel();
							        if (connectCompleteHandler != null)
								        connectCompleteHandler.handle(future);
						        }
					        });
				}

			} catch (InterruptedException e) {
				channel = null;
			}
		} finally {
			lock.unlock();
		}
	}

	public void send(byte[] data) {
		send(data, null);
	}

	public void send(byte[] data,
	        final FutureCompleteHandler sendCompleteHandler) {
		lock.lock();
		try {
			if (channel == null)
				return;
			ByteBuf buffer = Unpooled.copiedBuffer(data);

			ChannelFuture future = channel.writeAndFlush(buffer);
			if (sendCompleteHandler == null) {
				try {
					future.sync();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			} else {
				future.addListener(new GenericFutureListener<ChannelFuture>() {

					@Override
					public void operationComplete(ChannelFuture future)
					        throws Exception {
						if (sendCompleteHandler != null)
							sendCompleteHandler.handle(future);
					}

				});
			}
		} finally {
			lock.unlock();
		}
	}

}
